//
// Created by ultcrt on 2021/5/19.
//

#include "redis_io.h"
#include <thread>

using namespace nlohmann;
using namespace sw::redis;
using namespace std;

RedisIO::RedisIO(char *redis_url): redis_server(redis_url) {}

nlohmann::json RedisIO::fetch(const vector<string>& keys) {
    vector<future<json>> json_futures;
    vector<thread> fetch_threads;

    for(const string & key : keys) {
        promise<json> json_promise;
        json_futures.emplace_back(json_promise.get_future());
        fetch_threads.emplace_back(
                &RedisIO::fetch_thread,
                this,
                key,
                move(json_promise)
        );
    }

    // May cause error if not detach or join
    for(thread& fetch_thread: fetch_threads) {
        fetch_thread.join();
    }

    json result;

    for(int idx=0; idx < keys.size(); idx++) {
        json fetch_json = json_futures[idx].get();
        string key = keys[idx];
        result.update({key, fetch_json});
    }

    return result;
}

void RedisIO::fetch_thread(const string& key, promise<json> json_promise) {
    vector<string> latest;
    redis_server.lrange("key", 0, -1, back_inserter(latest));
    json result = {};

    if(latest.empty()) {
        json_promise.set_value(result);
    }
    else {
        for(const string& row: latest) {
            result.update(json::parse(row));
        }
    }
}

void RedisIO::push(const std::string& key, const nlohmann::json& val, int max_len) {
    redis_server.rpush(key, val.dump());

    vector<string> redis_list;
    redis_server.lrange(key, 0, -1, back_inserter(redis_list));

    if(redis_list.size() > max_len) {
        redis_server.blpop(key);
    }
}
